package com.easydata.agent.service;

import com.easydata.agent.core.DAG;
import com.easydata.agent.core.DAGContext;
import com.easydata.agent.core.dataIO.FileCacheDataReader;
import com.easydata.agent.core.dataIO.FileCacheDataWriter;
import com.easydata.agent.core.entity.TaskDefinition;
import com.easydata.agent.core.entity.TaskNode;
import com.easydata.agent.core.entity.TaskNodeLink;
import com.easydata.agent.exceptions.TaskException;
import com.easydata.libPlugin.core.AbstractDAGNode;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;

@Service
public class DAGService {

    @Autowired
    PluginService pluginService;

    public DAG buildDAG(DAGContext context, TaskDefinition dagDefinition) throws TaskException {
        var dag = new DAG();
        try {

            for (TaskNode n : dagDefinition.getNodes()) {
                AbstractDAGNode node = pluginService.getByName(n.getPluginName());
                node.setId(n.getId());
//                node.setParameter(n.getParameter());
                node.setReader(new FileCacheDataReader(context, node.getId()));
                node.setWriter(new FileCacheDataWriter(context));
                node.init(context, n.getParameter());
                dag.addNode(node);
            }

            for (TaskNodeLink l : dagDefinition.getLinks()) {
                dag.link(l.getFrom(), l.getTo());
                FileCacheDataWriter writer = (FileCacheDataWriter) dag.getNode(l.getFrom()).getWriter();
                writer.addDestination(l.getTo());
            }
        } catch (Exception ex) {

            throw new TaskException(-1, "failed to create DAG: " + ex.getMessage());
        }
        return dag;
    }

    public List<AbstractDAGNode> findEntry(DAG dag) {

        List<AbstractDAGNode> nodes = dag.getNodes();

        return nodes.stream().filter(n -> dag.getForwardNodes(n.getId()).isEmpty()).toList();
    }
}
